package com.mangix.core;

import com.mangix.conf.MangixConfig;
import com.mangix.db.MangixDB;
import com.mangix.files.MangixFiles;
import com.mangix.http.MangixHttpClient;
import com.mangix.http.MangixHttpHeader;
import com.mangix.scheduler.MangixScheduler;
import com.mangix.utils.MangixUtils;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.util.StringContentProvider;
import org.eclipse.jetty.http.HttpMethod;
import org.json.simple.JSONArray;
import org.json.simple.JSONValue;

import java.io.IOException;
import java.nio.file.Path;
import java.util.*;


/**
 * Created by Ahan on 2014/5/13.
 */
public class MangixNode {

    private static long MAX_JUMP_NUM;
    private static int M;
    private static long RING_SIZE;
    private static Map<String, MangixNode> localNodes;
    private static MangixConfig config;

    private String hostname;
    private int port;
    private long id;
    private MangixDB mangixDB;
    private MangixFiles mangixFiles;
    private MangixMonitor monitor;
    private boolean isLocal = false;
    protected ArrayList<MangixNode> successorList;
    protected ArrayList<MangixNode> predecessorList;
    private int successorListSize;
    private int predecessorListSize;
    private int nextFingerEntry = 0;
    private MangixScheduler scheduler;
    private long lastUpdateTime;


    /**
     * Create a remote MangixNode.
     * @param hostname
     * @param port
     */
    private MangixNode(String hostname, int port) throws Exception {
        this(hostname, port, false);
    }

    /**
     * Create a MangixNode.
     * @param hostname
     * @param port
     */
    private MangixNode(String hostname, int port, boolean isLocal) throws Exception {
        this.hostname = hostname;
        this.port = port;
        this.isLocal = isLocal;
        this.id = MangixUtils.hash(hostname + ":" + port, M);

        if (this.isLocal) {
            this.successorListSize = config.getMaxReplicasNum();
            this.predecessorListSize = config.getMaxReplicasNum();
            successorList = new ArrayList<>();
            predecessorList = new ArrayList<>();
            this.lastUpdateTime = System.currentTimeMillis();
        }

    }

    /**
     * Create all local nodes. Make sure this function is called before create any node.
     * @param config
     */
    public synchronized static void createLocalNodes(MangixConfig config) throws Exception {
        if (MangixNode.config != null) {
            return;
        }
        MangixNode.config = config;
        M = config.getM();
        RING_SIZE = (long) Math.pow(2, M);
        MAX_JUMP_NUM = RING_SIZE;
        // Create the local nodes.
        int[] ports = config.getPorts();
        localNodes = new HashMap<>();
        for (int port : ports) {
            MangixNode localNode = null;
            try {
                localNode = new MangixNode(config.getHostName(), port, true);
            } catch (Exception e) {
                e.printStackTrace();
                System.exit(-1);
            }
            localNodes.put(localNode.getHostName() + ":" + localNode.getPort(), localNode);
            System.out.println("Add Local Node: " + localNode.getHostName() + ":" + localNode.getPort());
        }
        // All the local nodes are created now. So we can init the DB and Files.
        // initDB and initFiles may call the function createNode, so they must be called after all the local nodes
        // are created.
        for (MangixNode node : getLocalNodeList()) {
            node.initDB();
            node.initFiles();
        }
    }

    public void initDB() throws Exception {
        if (this.mangixDB == null) {
            this.mangixDB = new MangixDB(config, port);
            mangixDB.init(this);
        }
    }

    public void initFiles() throws Exception {
        if (this.mangixFiles == null) {
            this.mangixFiles = new MangixFiles(port);
        }
    }
    /**
     * Create a Mangix node instance. If the node which is specified by hostname and port is a local node,
     * this function will always return the same node. Otherwise this function will create a new remote node.
     * This function must be called after the createLocalNodes(MangixConfig config) function, otherwise it will always return
     * null.
     * @param hostname
     * @param port
     * @return
     */
    public static MangixNode createNode(String hostname, int port) {
        if (config == null || localNodes == null) {
            return null;
        }
        if (hostname == "null" && port == 0)
            return null;
        MangixNode node = localNodes.get(hostname + ":" + port);
        if (node == null) {
            try{
                return new MangixNode(hostname, port);
            } catch (Exception e) {
                e.printStackTrace();
                return null;
            }
        } else {
            return node;
        }

    }

    /**
     * Get a list of local nodes.
     * @return
     */
     public synchronized static List<MangixNode> getLocalNodeList() {
         List<MangixNode> nodeList = new ArrayList<MangixNode>();
         if (localNodes != null) {
             for (MangixNode node: localNodes.values()) {
                 if (node.isLocal()) {
                     nodeList.add(node);
                 }
             }
         }
         return  nodeList;
    }

    public long hash() {
        return this.id;
    }

    public String getHostName() {
        return this.hostname;
    }

    public int getPort() {
        return this.port;
    }

    public MangixConfig getConfig() {
        return this.config;
    }

    public MangixDB getDB() {
        return this.mangixDB;
    }

    public MangixFiles getFiles() {
        return this.mangixFiles;
    }

    public boolean isLocal() {
        return this.isLocal;
    }

    public void init() throws Exception {
        init(null);
    }

    private void initSelf() throws Exception {
        System.out.println("Initialize self as a single node");
        mangixDB.initFingerTable(this, false);
        mangixDB.updatePredecessor(this);
        schedule();
    }

    public void init(MangixNode knownNode) throws Exception {
        if (knownNode == null) {
            System.out.println("Node " + this.toString() + " begins to initialize.");
            // This mangix node is the only one node in cluster.
            List<MangixNode> knownNodes = new ArrayList<>();
            Finger[] fingers = mangixDB.getFingerTable();
            for (Finger finger : fingers) {
                MangixNode node = finger.node;
                if (node!=null && !node.isLocal() && node.isActive()) {
                    knownNodes.add(node);
                }
            }
            if (knownNodes.size() != 0) {
                boolean isInitSuccess = false;
                for (MangixNode node : knownNodes) {
                    try {
                        init(node);
                        isInitSuccess = true;
                        break;
                    } catch (Exception e) {
                        // Do nothing.
                    }
                }
                if (!isInitSuccess) {
                    // This node can not initialize from any other nodes. Treat it as a single node.
                    initSelf();
                }
            } else {
                // This node only know itself.
                initSelf();
            }
            System.out.println("mangix.Mangix node " + this.toString() + " finishes initializing.");
        } else {
            join(knownNode);
            schedule();
        }

    }

    /**
     * Schedules some specified tasks for repeated fixed-delay execution.
     */
    private void schedule() throws Exception {
        if (scheduler == null) {
            scheduler = new MangixScheduler(this);
            scheduler.start();
        }
    }

    /**
     * Try to join the Mangix network. If success, the successor will be updated, otherwise the finger table will remain unchanged.
     * @param node
     * @throws Exception
     */
    public void join(MangixNode node) throws Exception {
        try {
            MangixNode successor = node.findSuccessor(this.hash());
            if (successor == null) {
                if (node.isActive()) {
                    mangixDB.updateFingerTable(0, node);
                }
            } else {
                mangixDB.updateFingerTable(0, successor);
            }
            update();
            System.out.println("Node " + this.toString() + " finishes initializing.");
        } catch(Exception e) {
            e.printStackTrace();
            if (node.isActive()) {
                mangixDB.updateFingerTable(0, node);
                update();
            }
        }
    }

    /**
     * Notify node target that the newPredecessor may be its new predecessor.
     *
     * @param newPredecessor The new predecessor node.
     */
    public void notify(MangixNode newPredecessor) throws Exception {
        if (isLocal) {
            MangixNode predecessor = getPredecessor();
            if (predecessor == null || !predecessor.isActive() || MangixUtils.isIDInInterval(newPredecessor.hash(), predecessor.hash(), this.hash(), false, false, M)) {
                updatePredecessor(newPredecessor);
            }
        } else {
            MangixHttpClient.put("http://" + hostname + ":" + port + "/predecessor", "{\"hostname\":\"" + newPredecessor.getHostName() + "\",\"port\":" + newPredecessor.getPort() + "}");
        }

    }

    /**
     * Ask node to find id's successor.
     *
     * @param id The object's hash id.
     * @return The id's successor. This value may be null.
     */
    public MangixNode findSuccessor(long id) throws Exception {
        MangixNode[] nodes = this.findPredecessor(id);
        return nodes[1];
    }

    /**
     * Ask node to find id's predecessor.
     *
     * @param id The object's hash id.
     * @return The first element of the return array is the predecessor of id, and the second is the successor of the first element.
     */
    public MangixNode[] findPredecessor(long id) throws Exception {
        MangixNode n = this;
        MangixNode nSuccessor = getSuccessor();
        long i = 0;
        while (!MangixUtils.isIDInInterval(id, n.hash(), nSuccessor.hash(), false, true, M) && i < MAX_JUMP_NUM) {
            MangixNode[] nodes = n.closetPrecedingFinger(id);
            // Avoid infinite loop.
            if (n.equals(nodes[0])) {
                return new MangixNode[]{n, nSuccessor};
            }
            n = nodes[0];
            nSuccessor = nodes[1];
            i ++;
        }
        if (i == MAX_JUMP_NUM) {
            System.out.println("The findPredecessor has reach the MAX JUMP NUM: " + MAX_JUMP_NUM + ".");
        }
        return new MangixNode[]{n, nSuccessor};
    }

    public MangixNode[] closetPrecedingFinger(long id) throws Exception {
        if (isLocal) {
            int m = config.getM();
            Finger[] fingerTable = mangixDB.getFingerTable();
            for (int i = m - 1; i >= 0; i--) {
                MangixNode tempNode = fingerTable[i].node;
                if (tempNode != null && MangixUtils.isIDInInterval(tempNode.hash(), this.hash(), id, false, false, m)) {
                    try {
                        MangixNode successor = tempNode.getSuccessor();
                        return new MangixNode[]{tempNode, successor};
                    } catch (Exception e) {
                        mangixDB.updateFingerTable(i, null);
                    }
                }
            }
            return new MangixNode[]{this, getSuccessor()};
        } else {
            String jsonString = MangixHttpClient.get("http://" + hostname + ":" + port + "/closet-preceding-finger?id=" + id);
            Map map = (Map) JSONValue.parse(jsonString);
            String hostname = map.get("hostname").toString();
            int port = Integer.parseInt(map.get("port").toString());
            Map successorInfo = (Map) map.get("successor");
            String successorHostname = successorInfo.get("hostname").toString();
            int successorPort = Integer.parseInt(successorInfo.get("port").toString());
            return new MangixNode[]{createNode(hostname, port), createNode(successorHostname, successorPort)};
        }
    }

    /**
     * Update the predecessor..
     *
     * @param newPredecessor The new predecessor.
     */
    private void updatePredecessor(MangixNode newPredecessor) throws Exception {
        if (isLocal) {
            mangixDB.updatePredecessor(newPredecessor);
            update();
        } else {
            MangixHttpClient.put("http://" + hostname + ":" + port + "/predecessor", "{\"hostname\":\"" + newPredecessor.getHostName() + "\",\"port\":" + newPredecessor.getPort() + "}");
        }
        System.out.println("Update node " + this.toString() + "'s predecessor to " + newPredecessor.toString() + ".");
    }

    /**
     * Get the successor.
     *
     * @return The successor of this node.
     */
    public MangixNode getSuccessor() throws Exception {
        if (isLocal) {
            return mangixDB.getSuccessor();
        } else {
            String jsonString = MangixHttpClient.get("http://" + hostname + ":" + port + "/successor");
            Map map = (Map) JSONValue.parse(jsonString);
            String hostname = map.get("hostname").toString();
            int port = Integer.parseInt(map.get("port").toString());
            return createNode(hostname, port);
           }
    }

    /**
     * Get the predecessor.
     * @return The predecessor of this node. If any exception occurs, return null.
     */
    public MangixNode getPredecessor() {
        if (isLocal) {
            return mangixDB.getPredecessor();
        } else {
            try {
                String jsonString = MangixHttpClient.get("http://" + hostname + ":" + port + "/predecessor");
                Map map = (Map) JSONValue.parse(jsonString);
                String hostname = map.get("hostname").toString();
                int port = Integer.parseInt(map.get("port").toString());
                return createNode(hostname, port);
            } catch (IOException e) {
                System.out.println("Http Error when getting " + this.toString() + "'s predecessor:" + e.getMessage());
                return null;
            }
        }
    }

    /**
     * Ensure tha the successor is correct.
     * @throws Exception
     */
    public void stabilize() throws Exception {
        MangixNode successor = getSuccessor();
        if (successor == null || !successor.isActive()) {
            // If this the successor of this node is down or can not be connected.
            while(successorList.size() > 0 && !successorList.get(0).isActive()) {
                successorList.remove(0);
            }
            if (successorList.size() == 0) {
                for (int i = M - 1; i >= 0; i--) {
                    if (mangixDB.getFinger(i).node != null && mangixDB.getFinger(i).node.isActive()) {
                        join(mangixDB.getFinger(i).node);
                        return;
                    }
                }
                // All the successors and the fingers are not alive.
                // Try to join the local nodes.
                List<MangixNode> localNodes = getLocalNodeList();
                System.out.println("All the successors and the fingers are not alive. Try to join the first local nodes.");
                join(localNodes.get(0));
                return;
            } else {
                mangixDB.updateFingerTable(0, successorList.get(0));
                update();
                return;
            }

        }
        // Update the successor.
        MangixNode predecessor = successor.getPredecessor();
        if (predecessor != null && predecessor.isActive() && !predecessor.equals(this) && predecessor.hash() == this.hash()) {
            System.out.println("This node (" + this.toString() + ", " + this.id + ") conflicts with node (" + predecessor.toString() + ", " + predecessor.hash()  + ") , they have the same hash id. Please modify the config file. Or you can increase the value of M.");
            System.exit(-1);
        }
        if (predecessor != null && predecessor.isActive() && MangixUtils.isIDInInterval(predecessor.hash(), this.hash(), successor.hash(), false, false, M)) {
            mangixDB.updateFingerTable(0, predecessor);
            System.out.println("Stabilize " + this.toString() + ", update successor to " + predecessor.toString());
            update();
        }
        // Notify the successor.
        successor = getSuccessor();
        successor.notify(this);
        // Update successor list.
        ArrayList<MangixNode> newSuccessorList = successor.getSuccessorList();
        successorList.clear();
        successorList.add(successor);
        if (newSuccessorList.size() < successorListSize)
            successorList.addAll(newSuccessorList);
        else
            successorList.addAll(newSuccessorList.subList(0, successorListSize - 1));
        // Update predecessor list.
        predecessor = getPredecessor();
        if (predecessor != null) {
            ArrayList<MangixNode> newPredecessorList = predecessor.getPredecessorList();
            predecessorList.clear();
            predecessorList.add(predecessor);
            if (newPredecessorList.size() < predecessorListSize)
                predecessorList.addAll(newPredecessorList);
            else
                predecessorList.addAll(newPredecessorList.subList(0, predecessorListSize - 1));
        }
    }

    public ArrayList<MangixNode> getSuccessorList() throws Exception {
        if (isLocal)
            return successorList;
        String jsonString = MangixHttpClient.get("http://" + hostname + ":" + port + "/successor-list");
        Map map = (Map) JSONValue.parse(jsonString);
        JSONArray successorListJson = (JSONArray) map.get("successor-list");
        ArrayList<MangixNode> successorList = new ArrayList<>();
        for (Object i : successorListJson) {
            String hostname = ((Map) i).get("hostname").toString();
            int port = Integer.parseInt(((Map) i).get("port").toString());
            successorList.add(createNode(hostname, port));
        }
        return successorList;
    }

    /**
     * Get the predecessor list of this node.
     * If any exception occurs, return an empty list.
     * @return
     */
    public ArrayList<MangixNode> getPredecessorList() {
        if (isLocal)
            return predecessorList;
        try {
            String jsonString = MangixHttpClient.get("http://" + hostname + ":" + port + "/predecessor-list");
            Map map = (Map) JSONValue.parse(jsonString);
            JSONArray predecessorListJson = (JSONArray) map.get("predecessor-list");
            ArrayList<MangixNode> ret = new ArrayList<>();
            for (Object i : predecessorListJson) {
                String hostname = ((Map) i).get("hostname").toString();
                int port = Integer.parseInt(((Map) i).get("port").toString());
                ret.add(createNode(hostname, port));
            }
            return ret;
        } catch (Exception e) {
            return new ArrayList<>();
        }
    }

    public boolean isActive() {
        try{
            MangixHttpClient.get("http://" + hostname + ":" + port + "/is-active");
            return true;
        } catch (Exception e) {
            System.out.println("Node " + hostname + ":" + port + " is not active.");
            return false;
        }
    }



    /**
     * Fix finger table one by one.
     */
    public void fixFingers() {
        for (int i = 1; i < M; i ++) {
            MangixNode newFinger = null;
            try{
                newFinger = findSuccessor(mangixDB.getFinger(i).start);
            } catch (Exception e) {
                e.printStackTrace();
            }
            try {
                if(newFinger != null && !newFinger.isActive()) {
                    newFinger = null;
                }
                if(newFinger == null) {
                    MangixNode nextFinger = mangixDB.getFinger((i + 1) % M).node;
                    if(nextFinger!=null && nextFinger.isActive()) {
                        mangixDB.updateFingerTable(i, nextFinger);
                        System.out.println("Update " + this.toString() + " " + i + "'s finger from null to " + nextFinger.toString());
                    } else {
                        mangixDB.updateFingerTable(i, newFinger);
                        System.out.println("Update " + this.toString() + " " + i + "'s finger to null");
                    }
                    update();
                } else {
                    if(!newFinger.equals(mangixDB.getFinger(i).node)) {
                        if (mangixDB.getFinger(i).node != null)
                            System.out.println("Update " + this.toString() + " " + i + "'s finger from " + mangixDB.getFinger(i).node.toString() + " to " + newFinger.toString());
                        else
                            System.out.println("Update " + this.toString() + " " + i + "'s finger from null to " + newFinger.toString());
                        mangixDB.updateFingerTable(i, newFinger);
                        update();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void setMonitor(MangixMonitor monitor) {
        this.monitor = monitor;
    }

    public MangixMonitor getMonitor() {
        return this.monitor;
    }

    public void update() {
        this.lastUpdateTime = System.currentTimeMillis();
        if (scheduler != null) {
            scheduler.update();
        }
    }

    public long getLastUpdateTime() {
        return this.lastUpdateTime;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj instanceof MangixNode) {
            MangixNode node = (MangixNode) obj;
            return node.getHostName().equals(this.hostname) && node.getPort() == this.port;
        } else {
            return false;
        }
    }

    @Override
    public int hashCode() {
        long id = MangixUtils.hash(this.hostname + ":" + this.port, config.getM());
        return (int) id;
    }

    @Override
    public String toString() {
        return hostname + ":" + port;
    }

    /**
     * Get the meta of this key from this node. If the key does not exist in this node, return null.
     * If any exception occurs, return null.
     * @param key
     * @return
     * @throws Exception
     */
    public Map<String, Object> getMeta(String key) throws Exception {
        if (isLocal) {
            Map<String, Object> meta = new HashMap();
            Path file = mangixFiles.getLatestFile(key);
            if (file == null) {
                return null;
            }
            String version = file.getFileName().toString().split("\\.")[0];
            meta.put("version", Long.parseLong(version));
            int replicasNum = mangixFiles.getReplicasNum(file);
            meta.put("replicas_num", replicasNum);
            List locations = new ArrayList<Map>();
            Map location = new HashMap();
            location.put("hostname", hostname);
            location.put("port", port);
            locations.add(location);
            for (int i = 0; i < replicasNum - 1; i ++) {
                MangixNode successor = successorList.get(i);
                location = new HashMap();
                location.put("hostname", successor.getHostName());
                location.put("port", successor.getPort());
                locations.add(location);
            }
            meta.put("locations", locations);
            return meta;
        } else {
            try {
                String jsonString = MangixHttpClient.head("http://" + hostname + ":" + port + "/direct/objects/" + key, MangixHttpHeader.META);
                Map meta = (Map) JSONValue.parse(jsonString);
                return meta;
            } catch (Exception e) {
                e.printStackTrace();
                return null;
            }
        }

    }
}